#include <iostream>

#include <gtest/gtest.h>

#include <logger.hpp>
#include "../tablet_server/sstable_block_cache.hpp"

using namespace xbase;

TEST(test_sstable_block_cache, manager)
{
	sstable_block_cache_manager::init_instance(5120);
	
	for(int i=0; i<10; ++i)
	{
		sstable_block_buffer buffer;
		buffer.prepare(1024);
		sstable_block_cache block = buffer.commit_and_checkout(0, i);
		ASSERT_FALSE(! block);
		ASSERT_TRUE(block.block_index().empty());
		std::size_t number = ((i < 5) ? (i + 1) : 5);
		ASSERT_EQ(number, sstable_block_cache_manager::instance().cached_block_number());
		ASSERT_EQ(number * 1024, sstable_block_cache_manager::instance().cached_block_size());
		ASSERT_EQ(5120 - 1024, sstable_block_cache_manager::instance().available_memory());
	}
	ASSERT_EQ(5120, sstable_block_cache_manager::instance().available_memory());

	{
		// 放不下这个块，但不会使已有的块淘汰
		{
			sstable_block_buffer buffer;
			buffer.prepare(10240);
			sstable_block_cache block = buffer.commit_and_checkout(2, 0);
			ASSERT_TRUE(! block);
			ASSERT_EQ(5, sstable_block_cache_manager::instance().cached_block_number());
			ASSERT_EQ(5 * 1024, sstable_block_cache_manager::instance().cached_block_size());
			ASSERT_EQ(5120, sstable_block_cache_manager::instance().available_memory());
		}
		
		sstable_block_cache block0(0, 0);
		ASSERT_TRUE(! block0);

		sstable_block_cache block9(0, 9);
		ASSERT_FALSE(! block9);

		sstable_block_buffer buffer;
		buffer.prepare(2050);
		ASSERT_FALSE(! buffer.commit(1, 0));

		ASSERT_EQ(3, sstable_block_cache_manager::instance().cached_block_number());
		ASSERT_EQ(2 * 1024 + 2050, sstable_block_cache_manager::instance().cached_block_size());
		ASSERT_EQ(5120 - 1024, sstable_block_cache_manager::instance().available_memory());

		std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;
	}

	sstable_block_cache_manager::instance().clear_sstable_cache(0);
	ASSERT_EQ(1, sstable_block_cache_manager::instance().cached_block_number());
	ASSERT_EQ(2050, sstable_block_cache_manager::instance().cached_block_size());
	ASSERT_EQ(5120, sstable_block_cache_manager::instance().available_memory());

	std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;

	sstable_block_cache_manager::reset_instance();
}

TEST(test_sstable_block_cache, holder)
{
	sstable_block_cache_manager::init_instance(5120);

	sstable_block_cache holder;
	ASSERT_EQ(0, holder.ref_count());
	for(int i=0; i<10; ++i)
	{
		sstable_block_buffer buffer;
		buffer.prepare(1024);
		sstable_block_cache block = buffer.commit_and_checkout(0, i);
		ASSERT_FALSE(! block);
		ASSERT_TRUE(block.block_index().empty());
		ASSERT_EQ(1, block.ref_count());
		holder = block;
		ASSERT_EQ(2, block.ref_count());
		ASSERT_EQ(1, block.node()->m_value.ref_count());
	}
	ASSERT_EQ(1, holder.ref_count());

	{
		cache_node *node = holder.node();
		holder.release();
		ASSERT_EQ(0, holder.ref_count());
		ASSERT_EQ(0, node->m_value.ref_count());
	}

	{
		sstable_block_cache another(0, 9);
		ASSERT_FALSE(! another);
		cache_node *node = sstable_block_cache_manager::instance().checkout(cache_key(0, 9));
		ASSERT_TRUE(node == another.node());
		
		ASSERT_EQ(1, another.ref_count());
		ASSERT_EQ(2, node->m_value.ref_count());
		
		{
			sstable_block_cache adopt(node, adopt_sstable_block_cache);
			sstable_block_cache copy(adopt);
			ASSERT_FALSE(! copy);
			ASSERT_EQ(2, adopt.ref_count());
			ASSERT_EQ(2, node->m_value.ref_count());
		}
		ASSERT_EQ(1, node->m_value.ref_count());
		sstable_block_cache guarder(node);
		ASSERT_FALSE(! guarder);
		ASSERT_EQ(2, node->m_value.ref_count());
		ASSERT_EQ(1, guarder.ref_count());

		another.release();
		ASSERT_EQ(1, node->m_value.ref_count());

		guarder.release();
		ASSERT_EQ(0, node->m_value.ref_count());
	}

	{
		sstable_block_cache empty(1, 9);
		ASSERT_TRUE(! empty);
		sstable_block_cache empty_copy(holder);
		ASSERT_TRUE(! empty_copy);
	}

	std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;
	sstable_block_cache_manager::reset_instance();
}

TEST(test_sstable_block_cache, buffer)
{
	sstable_block_cache_manager::init_instance(5120);

	{
		sstable_block_buffer buffer;
		ASSERT_TRUE(! buffer);
		boost::asio::mutable_buffer buf = buffer.prepare(1024);
		ASSERT_FALSE(! buffer);
		ASSERT_EQ(1024, boost::asio::buffer_size(buf));
		buffer.commit(1024);

		ASSERT_TRUE(buffer.commit(0, 0));
		ASSERT_TRUE(! buffer);
		ASSERT_FALSE(buffer.commit(0, 0));
		ASSERT_FALSE(buffer.commit(0, 1));
		ASSERT_FALSE(buffer.commit_and_checkout(0, 0));
		ASSERT_FALSE(buffer.commit_and_checkout(0, 2));

		sstable_block_cache block(0, 0);
		ASSERT_FALSE(! block);
		ASSERT_TRUE(block.block_index().empty());
	}

	{
		sstable_block_buffer buffer;
		ASSERT_FALSE(buffer.commit(1, 0));
		buffer.prepare(2048);
		ASSERT_FALSE(buffer.commit(0, 0)); // collide

		ASSERT_FALSE(! buffer); // 数据仍在
		ASSERT_FALSE(! buffer.commit_and_checkout(1, 0)); // 可成功提交
	}

	{
		sstable_block_buffer buffer;
		buffer.prepare(10240);
		ASSERT_TRUE(! buffer.commit_and_checkout(2, 0)); // to big
		ASSERT_FALSE(! buffer); // 数据仍在
		buffer.release();
		ASSERT_TRUE(! buffer); // 数据释放了
	}

	{
		sstable_block_buffer buffer;
		ASSERT_TRUE(! buffer.commit_and_checkout(3, 0)); // no data
		ASSERT_TRUE(! buffer.commit(3, 0)); // no data
		ASSERT_TRUE(! buffer);
	}

	std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;
	sstable_block_cache_manager::reset_instance();
}
